package com.gxlevi.task

import com.gxlevi.bean.ClickLogWide
import com.gxlevi.util.HBaseUtil
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, WindowedStream}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.sink.SinkFunction

case class ChannelPvUv(
                        var channelId: String,
                        var yearMonthDayHour: String,
                        var pv: Long,
                        var uv: Long
                      )

object ChannelPvUvTask extends BaseTask[ChannelPvUv] {
  override def map(clickWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelPvUv] = {
    clickWideDataStream.flatMap {
      clickLogWide =>
        List(
          ChannelPvUv(clickLogWide.categoryID, clickLogWide.yearMonthDayHour, clickLogWide.count, clickLogWide.isHourNew),
          ChannelPvUv(clickLogWide.categoryID, clickLogWide.yearMonthDay, clickLogWide.count, clickLogWide.isDayNew),
          ChannelPvUv(clickLogWide.categoryID, clickLogWide.yearMonth, clickLogWide.count, clickLogWide.isMonthNew)
        )
    }
  }

  override def keyBy(mapStream: DataStream[ChannelPvUv]): KeyedStream[ChannelPvUv, String] = {
    mapStream.keyBy {
      channelPvUv => channelPvUv.channelId + channelPvUv.yearMonthDayHour
    }
  }

  override def reduce(windowedStream: WindowedStream[ChannelPvUv, String, TimeWindow]): DataStream[ChannelPvUv] = {
    windowedStream.reduce {
      (t1, t2) =>
        ChannelPvUv(t1.channelId, t1.yearMonthDayHour, t1.pv + t2.pv, t1.uv + t2.uv)
    }
  }

  override def sinkToHBase(reduceDataStream: DataStream[ChannelPvUv]): Unit = {
    reduceDataStream.addSink(new SinkFunction[ChannelPvUv] {
      override def invoke(value: ChannelPvUv): Unit = {
        tableName = "channel_pvuv"
        clfName = "info"
        val yearMonthDayHourColName = "yearMonthDayHour"
        rowKey = value.channelId + ":" + value.yearMonthDayHour

        val map = HBaseUtil.getMapData(tableName, rowKey, clfName, List(pvColumn, uvColumn))
        val totalPV = getTotal(map, pvColumn, value.pv)
        val totalUV = getTotal(map, uvColumn, value.uv)

        HBaseUtil.putMapData(tableName, rowKey, clfName, Map(
          channelIdColumn -> value.channelId,
          yearMonthDayHourColName -> value.yearMonthDayHour,
          pvColumn -> totalPV.toString,
          uvColumn -> totalUV.toString
        ))
      }
    })
  }
}
